#include "../common/dispatch.hpp"
#include "requestor.hpp"
#include "rpc_caller.hpp"
#include "rpc_registry.hpp"
#include "rpc_topic.hpp"


namespace rpc
{
    namespace client
    {
        class RegistryClient
        {
        public: 
            using ptr = std::shared_ptr<RegistryClient>;
            RegistryClient(const std::string& ip ,const int& port) 
            : _requestor(std::make_shared<Requestor>()),
            _dispatcher(std::make_shared<Dispatcher>()),
            _provider(std::make_shared<Provider>(_requestor))
            {
                auto rsp_cb = std::bind(&Requestor::onResponse,_requestor.get(),std::placeholders::_1,std::placeholders::_2);
               _dispatcher->registerHandler<BaseMessage>(Mtype::RSP_SERVICE,rsp_cb);

               auto msg_cb = std::bind(&Dispatcher::onMessage,_dispatcher.get(),std::placeholders::_1,std::placeholders::_2);
               _client = CilentFactory::create(ip,port);
               _client->setMessageCallback(msg_cb);
               _client->connect();
            }
            bool registryMethod(const std::string& method,const Address& host)
            {
                return _provider->registryMethod(_client->connection(),method,host);
            }
        private:
            Requestor::ptr _requestor;
            Provider::ptr _provider;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };

        class DiscoverClient
        {
        public:
            using ptr = std::shared_ptr<DiscoverClient>;
            DiscoverClient(const std::string& ip ,const int& port, const Discoverer::OfflineCallback &cb) 
            : _requestor(std::make_shared<Requestor>()),
            _dispatcher(std::make_shared<Dispatcher>()),
            _discoverer(std::make_shared<Discoverer>(_requestor,cb))
            {
                auto rsp_cb = std::bind(&Requestor::onResponse,_requestor.get(),std::placeholders::_1,std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(Mtype::RSP_SERVICE,rsp_cb);
                
                auto req_cb = std::bind(&Discoverer::onServiceRequest,_discoverer.get(),std::placeholders::_1,std::placeholders::_2);
                _dispatcher->registerHandler<ServiceRequest>(Mtype::REQ_SERVICE,req_cb);

                auto msg_cb = std::bind(&Dispatcher::onMessage,_dispatcher.get(),std::placeholders::_1,std::placeholders::_2);
                _client->setMessageCallback(msg_cb);
                _client->connect();
            }
            bool serviceDiscover(const std::string &method, Address &host)
            {
                return _discoverer->serviceDiscover(_client->connection(),method,host);//获取提供服务的主机
            }
        private:
            Dispatcher::ptr _dispatcher;
            Requestor::ptr _requestor;
            Discoverer::ptr _discoverer;
            BaseClient::ptr _client;
        };

        class RpcClient
        {
        public:
            using ptr = std::shared_ptr<RpcClient>;
            RpcClient(bool enableDiscovery,const std::string& ip,const int& port) 
            : _enableDiscovery(enableDiscovery),
            _requestor(std::make_shared<Requestor>()),
            _dispatcher(std::make_shared<Dispatcher>()),
            _caller(std::make_shared<rpc::client::RpcCaller>(_requestor))
            {
                //针对rpc请求后的响应进行的回调处理
                auto rsp_cb = std::bind(&client::Requestor::onResponse,_requestor.get(),
                                        std::placeholders::_1,std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(Mtype::RSP_RPC,rsp_cb);

                //如果启用了服务发现，地址信息是注册中心的地址，是服务发现客户端需要连接的地址，则通过地址信息实例化discovery_client
                //如果没有启用服务发现，则地址信息是服务提供者的地址，则直接实例化好rpc_client
                if(_enableDiscovery)
                {
                    auto offline_cb = std::bind(&RpcClient::delClient, this,std::placeholders::_1);
                    _discover_client = std::make_shared<DiscoverClient> (ip,port,offline_cb);
                }
                else
                {
                    auto message_cb = std::bind(&Dispatcher::onMessage,_dispatcher.get(),std::placeholders::_1,std::placeholders::_2);
                    _rpc_client = CilentFactory::create(ip,port);
                    _rpc_client->setMessageCallback(message_cb);
                    _rpc_client->connect();
                }
            }
            bool call(const std::string& method,const Json::Value& params,Json::Value& result)
            {
                BaseClient::ptr client = getClient(method);
                if(client.get() == nullptr)
                {
                    return false;
                }
                return _caller->call(client->connection(),method,params,result);
            }
            bool call(const std::string& method,const Json::Value& params,RpcCaller::JsonAsyncResponse& result)
            {
                BaseClient::ptr client = getClient(method);
                if(client.get() == nullptr)
                {
                    return false;
                }
                return _caller->call(client->connection(),method,params,result);
            }
            bool call(const std::string& method,const Json::Value& params,const RpcCaller::JsonResponseCallback& result)
            {
                BaseClient::ptr client = getClient(method);
                if(client.get() == nullptr)
                {
                    return false;
                }
                return _caller->call(client->connection(),method,params,result);
            }
        private:
            void delClient(const Address& host)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _rpc_clients.erase(host);
            }
            BaseClient::ptr getClient(const Address& host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _rpc_clients.find(host);
                if (it == _rpc_clients.end())
                {
                    return BaseClient::ptr();
                }
                return it->second;
            }
            void putClient(const Address &host, BaseClient::ptr &client) 
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.insert(std::make_pair(host, client));
            }
            BaseClient::ptr newClient(const Address &host) 
            {
                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                auto client = CilentFactory::create(host.first, host.second);
                client->setMessageCallback(message_cb);
                client->connect();
                putClient(host, client);
                return client;
            }
            BaseClient::ptr getClient(const std::string& method)
            {
                BaseClient::ptr client;
                if(_enableDiscovery)
                {
                    //1. 通过服务发现，获取服务提供者地址信息
                    Address host;
                    bool ret = _discover_client->serviceDiscover(method,host);
                    if(ret == false)
                    {
                        ELOG("当前 %s 服务，没有找到服务提供者！", method.c_str());
                        return BaseClient::ptr();
                    }
                    //2. 查看服务提供者是否已有实例化客户端，有则直接使用，没有则创建
                    client = getClient(host);
                    if(client.get() == nullptr)
                    {
                        client = newClient(host);
                    }
                }
                else
                {
                    client = _rpc_client;
                }
                return client;
            }
        private:
            struct AddressHash
            {
                size_t operator()(const Address &host) const 
                {
                    std::string addr = host.first + std::to_string(host.second);
                    return std::hash<std::string>()(addr);
                }
            };
            
            bool _enableDiscovery;
            DiscoverClient::ptr _discover_client;
            Requestor::ptr _requestor;
            RpcCaller::ptr _caller;
            Dispatcher::ptr _dispatcher;
            std::mutex _mutex;
            BaseClient::ptr _rpc_client;
            std::unordered_map<Address,BaseClient::ptr,AddressHash> _rpc_clients;
        };

        class TopicClient
        {
        public:
            using ptr = std::shared_ptr<TopicClient>;
            TopicClient(const std::string &ip, int port) 
            : _requestor(std::make_shared<Requestor>()),
            _dispatcher(std::make_shared<Dispatcher>()),
            _topic_manager(std::make_shared<TopicManager>(_requestor))
            {
                auto rsp_cb = std::bind(&Requestor::onResponse,_requestor.get(),std::placeholders::_1,std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(Mtype::RSP_TOPIC,rsp_cb);

                auto msg_cb = std::bind(&TopicManager::onPublish, _topic_manager.get(),std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<TopicRequest>(Mtype::REQ_TOPIC,msg_cb);

                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _rpc_client = CilentFactory::create(ip,port);
                _rpc_client->setMessageCallback(message_cb);
                _rpc_client->connect();
            }
            bool create(const std::string &key) 
            {
                return _topic_manager->create(_rpc_client->connection(), key);
            }
            bool remove(const std::string &key) 
            {
                return _topic_manager->remove(_rpc_client->connection(), key);
            }
            bool subscribe(const std::string &key, const TopicManager::SubCallback &cb) 
            {
                return _topic_manager->subscribe(_rpc_client->connection(), key, cb);
            }
            bool cancel(const std::string &key) 
            {
                return _topic_manager->cancel(_rpc_client->connection(), key);
            }
            bool publish(const std::string &key, const std::string &msg) 
            {
                return _topic_manager->publish(_rpc_client->connection(), key, msg);
            }
            void shutdown() 
            {
                _rpc_client->shutdown();
            }
        private:
            Requestor::ptr _requestor;
            TopicManager::ptr _topic_manager;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _rpc_client;
        };
    }
}
